package jets

import (
	"context"
	"fmt"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
	"go.uber.org/zap"
	"sync"
	"time"
	"wzgames/internal/constant"
	"wzgames/internal/gloader"
	"wzgames/pkg/logger"
)

var (
	_mgr *NatsJets
	once sync.Once
)

type NatsJets struct {
	mu        sync.RWMutex
	natsC     *nats.Conn
	jetStream jetstream.JetStream

	streams   map[string]jetstream.Stream
	consumers map[string]jetstream.Consumer

	ctx    context.Context
	cancel context.CancelFunc
}

func NatsJetsMgr() *NatsJets {
	if _mgr == nil {
		once.Do(func() {
			_mgr = &NatsJets{
				streams:   make(map[string]jetstream.Stream),
				consumers: make(map[string]jetstream.Consumer),
			}
		})
	}

	return _mgr
}

func (nj *NatsJets) Init() error {
	conn, err := gloader.CommMgr().CreateNats()
	if err != nil {
		//logger.Error("err == ", zap.String("error", err.Error()))
		return err
	}
	nj.natsC = conn
	nj.jetStream, err = jetstream.New(nj.natsC)
	if err != nil {
		logger.Error("[JetStream] create jet stream fail", zap.Error(err))
		return err
	}
	nj.ctx, nj.cancel = context.WithCancel(context.Background())

	return nil
}

func (nj *NatsJets) CreateStream(cfg jetstream.StreamConfig) (jetstream.Stream, error) {
	nj.mu.Lock()
	defer nj.mu.Unlock()
	_, ok := nj.streams[cfg.Name]
	if ok {
		newS, err := nj.jetStream.UpdateStream(nj.ctx, cfg)
		if err != nil {
			logger.Error("[JetStream] create jet stream fail on UpdateStream", zap.String("Name", cfg.Name), zap.Error(err))
			return nil, err
		}
		nj.streams[cfg.Name] = newS
		return newS, nil
	}
	s1, e := nj.jetStream.CreateStream(nj.ctx, cfg)
	if e != nil {
		logger.Error("[JetStream] create jet stream fail on CreateStream", zap.String("Name", cfg.Name), zap.Error(e))
		return nil, e
	}
	nj.streams[cfg.Name] = s1

	return s1, nil
}

func (nj *NatsJets) CreateConsumer(streamName string, cfg jetstream.ConsumerConfig) (jetstream.Consumer, error) {
	nj.mu.Lock()
	defer nj.mu.Unlock()
	stream, ok := nj.streams[streamName]
	if !ok {
		logger.Error("[JetStream] create consumer fail on GetStream", zap.String("Steam", streamName), zap.String("Consumer", cfg.Name))
		return nil, fmt.Errorf("[JetStream] stream %s not found", streamName)
	}
	old, ok1 := nj.consumers[cfg.Name]
	if ok1 {
		return old, nil
	}
	c, e := stream.CreateOrUpdateConsumer(nj.ctx, cfg)
	if e != nil {
		logger.Error("[JetStream] create consumer fail on AddConsumer", zap.String("Steam", streamName), zap.String("Consumer", cfg.Name), zap.Error(e))
		return nil, e
	}
	nj.consumers[cfg.Name] = c

	return c, nil
}

func Publish(subject string, data []byte) bool {
	_, err := _mgr.jetStream.Publish(_mgr.ctx, subject, data)
	if err != nil {
		logger.Error("[JetStream] publish fail", zap.String("Subject", subject), zap.Error(err))
		return false
	}

	return true
}

func PublishAsync(subject string, data []byte) bool {
	_, err := _mgr.jetStream.PublishAsync(subject, data)
	if err != nil {
		logger.Error("[JetStream] publish async fail", zap.String("Subject", subject), zap.Error(err))
		return false
	}

	return true
}

func (nj *NatsJets) Destroy() {
	if nj.natsC != nil {
		nj.natsC.Drain()
	}
}

func (nj *NatsJets) CreateBetStream() (jetstream.Stream, error) {
	return nj.CreateStream(jetstream.StreamConfig{
		Name:     constant.StreamG8,
		Subjects: []string{constant.SubjectBetsWild},
		Discard:  jetstream.DiscardNew,
		MaxAge:   60 * 60 * 24 * 30 * time.Second,
	})
}

func (nj *NatsJets) CreateAvtBetStream() (jetstream.Stream, error) {
	return nj.CreateStream(jetstream.StreamConfig{
		Name:     constant.StreamAviator,
		Subjects: []string{constant.SubjectAvtWild},
		Discard:  jetstream.DiscardNew,
		MaxAge:   60 * 60 * 24 * 30 * time.Second,
	})
}

func (nj *NatsJets) CreateSettleStream() (jetstream.Stream, error) {
	return nj.CreateStream(jetstream.StreamConfig{
		Name:     constant.StreamSettle,
		Subjects: []string{constant.SubjectSettleWild},
		Discard:  jetstream.DiscardNew,
		MaxAge:   60 * 60 * 24 * 10 * time.Second,
	})
}

func (nj *NatsJets) InitStreams() error {
	_, err := nj.CreateBetStream()
	if err == nil {
		_, err = nj.CreateAvtBetStream()
	}
	if err == nil {
		_, err = nj.CreateSettleStream()
	}
	return err
}

func (nj *NatsJets) GetStream(streamName string) jetstream.Stream {
	nj.mu.RLock()
	defer nj.mu.RUnlock()
	stream, ok := nj.streams[streamName]
	if ok {
		return stream
	}
	return nil
}

func (nj *NatsJets) GetConsumer(name string) jetstream.Consumer {
	nj.mu.RLock()
	defer nj.mu.RUnlock()
	c, ok := nj.consumers[name]
	if ok {
		return c
	}
	return nil
}
